/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package remoting;

import common.constants.CommonConstants;
import common.exception.RemotingException;
import common.model.Codec2;
import common.model.Endpoint;
import common.model.MagicUrl;
import remoting.channel.Channel;
import remoting.channel.ChannelHandler;
import remoting.channel.ChannelHandlerDelegate;
import serialization.Serializer;

/**
 * AbstractPeer
 */
public abstract class AbstractPeer implements Endpoint, ChannelHandler {

  private final ChannelHandler handler;

  private volatile MagicUrl url;

  // closing closed means the process is being closed and close is finished
  private volatile boolean closing;

  private volatile boolean closed;

  private int timeout;

  private int connectTimeout;

  private Codec2 codec;

  public AbstractPeer(MagicUrl url, ChannelHandler handler, Codec2 codec) {
    if (url == null) {
      throw new IllegalArgumentException("url == null");
    }
    if (handler == null) {
      throw new IllegalArgumentException("handler == null");
    }
    this.url = url;
    this.handler = handler;
    this.codec = codec;
    this.timeout = url.getPositiveParameter(CommonConstants.TIMEOUT_KEY, CommonConstants.DEFAULT_TIMEOUT);
    this.connectTimeout = url.getPositiveParameter(CommonConstants.CONNECT_TIMEOUT_KEY, CommonConstants.DEFAULT_CONNECT_TIMEOUT);
  }

  @Override
  public void sendMessage(Object message) throws RemotingException {
    sendMessage(message, url.getParameter(CommonConstants.SENT_KEY, false));
  }

  @Override
  public void close() {
    closed = true;
  }

  @Override
  public void close(int timeout) {
    close();
  }

  @Override
  public void startClose() {
    if (isClosed()) {
      return;
    }
    closing = true;
  }

  @Override
  public MagicUrl getUrl() {
    return url;
  }

  protected void setUrl(MagicUrl url) {
    if (url == null) {
      throw new IllegalArgumentException("url == null");
    }
    this.url = url;
  }

  @Override
  public ChannelHandler getChannelHandler() {
    if (handler instanceof ChannelHandlerDelegate) {
      return ((ChannelHandlerDelegate) handler).getHandler();
    } else {
      return handler;
    }
  }

  /**
   * @return ChannelHandler
   */
  @Deprecated
  public ChannelHandler getHandler() {
    return getDelegateHandler();
  }

  /**
   * Return the final handler (which may have been wrapped). This method should be distinguished
   * with getChannelHandler() method
   *
   * @return ChannelHandler
   */
  public ChannelHandler getDelegateHandler() {
    return handler;
  }

  public Codec2 getCodec() {
    return codec;
  }

  @Override
  public boolean isClosed() {
    return closed;
  }

  public boolean isClosing() {
    return closing && !closed;
  }

  @Override
  public void connected(Channel ch) throws RemotingException {
    if (closed) {
      return;
    }
    handler.connected(ch);
  }

  @Override
  public void disconnected(Channel ch) throws RemotingException {
    handler.disconnected(ch);
  }

  @Override
  public void sent(Channel ch, Object msg) throws RemotingException {
    if (closed) {
      return;
    }
    handler.sent(ch, msg);
  }

  @Override
  public void received(Channel ch, Object msg) throws RemotingException {
    if (closed) {
      return;
    }
    handler.received(ch, msg);
  }

  @Override
  public void caught(Channel ch, Throwable ex) throws RemotingException {
    handler.caught(ch, ex);
  }

  protected int getTimeout() {
    return timeout;
  }

  protected int getConnectTimeout() {
    return connectTimeout;
  }

}
